package com.sunzm.flink.datastream.scala.robotcall

import com.alibaba.fastjson.JSON
import com.sunzm.flink.datastream.scala.beans.{RobotCall, RobotCallFeeResult}
import org.apache.flink.api.common.state.{BroadcastState, ReadOnlyBroadcastState}
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

class RobotCallFeeKeyedBroadcastProcessFunction extends KeyedBroadcastProcessFunction[String, RobotCall,
  String, RobotCallFeeResult] {

  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))

  override def processElement(value: RobotCall,
                              ctx: KeyedBroadcastProcessFunction[String, RobotCall, String, RobotCallFeeResult]#ReadOnlyContext,
                              out: Collector[RobotCallFeeResult]): Unit = {
    val bdState: ReadOnlyBroadcastState[Int, (Double, Double, Double, Double)] = ctx.getBroadcastState(RobotCallFeeJob.userStateDescriptor)

    val numberType: Int = value.numberType

    val direction: Int = value.direction

    val callWay: Int = value.callWay
    //计费时长
    val feeDuration: Int = value.feeDuration

    //初始化一个费率为 0, 单位分
    var feeRate: Int = 0

    if (bdState.contains(numberType)) {
      val (sipCallinCharge: Double, sipCalloutCharge: Double, phoneCallinCharge: Double, phoneCalloutCharge: Double) = bdState.get(numberType)

      (direction, callWay) match {
        case (0, 2) => {
          feeRate = (sipCallinCharge * 100).toInt
        }
        case (0, 3) => {
          feeRate = (phoneCallinCharge * 100).toInt
        }
        case (1, 2) => {
          feeRate = (sipCalloutCharge * 100).toInt
        }
        case (1, 3) => {
          feeRate = (phoneCalloutCharge * 100).toInt
        }
        case _ => {}
      }
    } else {
      logger.debug("numberType：{}, 没有对应的费率", numberType)
    }

    if (feeRate > 0) {
      //计算费用 单位：分
      val fee: Int = feeDuration * feeRate

      val companyId: String = value.companyId
      val startDate: String = value.startDate

      val robotCallFeeResult = RobotCallFeeResult(value.callID, startDate, companyId, numberType, fee, feeRate, feeDuration)

      out.collect(robotCallFeeResult)
    }

  }

  override def processBroadcastElement(value: String,
                                       ctx: KeyedBroadcastProcessFunction[String, RobotCall, String, RobotCallFeeResult]#Context,
                                       out: Collector[RobotCallFeeResult]): Unit = {

    logger.debug(s"收到费率信息: ${value}")

    val jSONObject = JSON.parseObject(value)

    val opType = jSONObject.getString("opType")

    val number_type = jSONObject.getIntValue("number_type")
    val sip_callin_charge = jSONObject.getDoubleValue("sip_callin_charge")
    val sip_callout_charge = jSONObject.getDoubleValue("sip_callout_charge")
    val phone_callin_charge = jSONObject.getDoubleValue("phone_callin_charge")
    val phone_callout_charge = jSONObject.getDoubleValue("phone_callout_charge")

    //获取广播状态
    val bdState: BroadcastState[Int, (Double, Double, Double, Double)] = ctx.getBroadcastState(RobotCallFeeJob.userStateDescriptor)

    opType match {
      case "CREATE" | "UPDATE" => {
        //添加 或者 更新
        logger.debug("费率发生添加或者更新操作")
        bdState.put(number_type, (sip_callin_charge, sip_callout_charge, phone_callin_charge, phone_callout_charge))
      }
      case "DELETE" => {
        //删除
        logger.debug("费率发生删除操作")
        bdState.remove(number_type)
      }
      case "READ" => {
        //读取
      }
      case _ => {
        logger.warn(s"不支持的操作类型: ${opType}")
      }
    }
  }
}
